package com.crawler.db

import java.sql.{PreparedStatement, Statement}
import java.util.Date

import com.crawler.message.TaskParamMessage
import com.crawler.tables.Tables._
import com.crawler.utils.MD5Utils
import org.slf4j.LoggerFactory

import scala.collection.mutable.ListBuffer

trait CrawlerTaskRepository{
  def insert(crawlerTask: TaskRow)
  def bulkInsert(tasks: List[TaskParamMessage], dbName: String)
  def find(crawlerTask: TaskRow): List[TaskParamMessage]
  def findByMd5(md5List: List[String], dbName: String): List[TaskParamMessage]
  def update(crawlerTask: TaskRow)
  def batchUpdate(crawlerTasks: List[TaskParamMessage])
  def delete(crawlerTask: TaskRow)
  def findByPrior(dbName:String): List[TaskParamMessage]
  def findTimeOutTask(dbName:String):List[TaskParamMessage]
  def findRunningTasks(dbName: String): Int
  def findTasksGroup(dbName: String): List[Tuple2[Int, Long]]
}

class H2CrawlerTaskRepository extends CrawlerTaskRepository{
  val log = LoggerFactory.getLogger(classOf[H2CrawlerTaskRepository])

  override def insert(crawlerTask: TaskRow): Unit  = {
    val isExsit = find(crawlerTask)
    if (isExsit.length > 0){
      return
    }

    val connectionPool = Datasource.getConnectionPool(crawlerTask.source).getConnection
    connectionPool.setAutoCommit(false)
    val sql = """create table IF NOT EXISTS "TASK"(
      "METHOD" VARCHAR NOT NULL ,
      "DATA" VARCHAR NOT NULL ,
      "HOST" VARCHAR NOT NULL ,
      "PARAM" VARCHAR NOT NULL ,
      "HEADER" VARCHAR NOT NULL ,
      "SOURCE" VARCHAR NOT NULL,
      "TYPE"  VARCHAR NOT NULL,
      "STATUS" INT NOT NULL,
      "PRIOR" INT NOT NULL,
      "DATE" BIGINT NOT NULL,
      "MD5" VARCHAR NOT NULL
    );
    CREATE INDEX IF NOT EXISTS ind_md5 ON  TASK(MD5);"""
    connectionPool.createStatement().execute(sql)

    var stmt: Statement = null
    try {
      stmt = connectionPool.createStatement()
      val insert =s"INSERT INTO task(method,host,data,param,header, source,type,status,prior,date,md5) VALUES ('${crawlerTask.method}', '${crawlerTask.host}','${crawlerTask.data}','${crawlerTask.param}','${crawlerTask.header}', '${crawlerTask.source}','${crawlerTask.`type`}','${crawlerTask.status}','${crawlerTask.prior}', '${new Date().getTime}','${MD5Utils.getMd5String((crawlerTask.host+crawlerTask.data+crawlerTask.param))}')"
      stmt.execute(insert)
      connectionPool.setAutoCommit(true)
    } catch {
      case e:Exception => e.printStackTrace()
    } finally {
      stmt.close()
      connectionPool.close()
    }
  }

  override def bulkInsert(tasks: List[TaskParamMessage], dbName: String): Unit = {
    val connectionPool = Datasource.getConnectionPool(dbName).getConnection
    connectionPool.setAutoCommit(false)
    var ps: PreparedStatement = null
    try {
      ps = connectionPool.prepareStatement("INSERT INTO task VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)")
      tasks.map(crawlerTask => {
        ps.setString(1,crawlerTask.method)
        ps.setString(2,crawlerTask.data)
        ps.setString(3,crawlerTask.host)
        ps.setString(4,crawlerTask.param)
        ps.setString(5,crawlerTask.header)
        ps.setString(6,crawlerTask.source)
        ps.setString(7,crawlerTask.`type`)
        ps.setInt(8,crawlerTask.status)
        ps.setInt(9,crawlerTask.prior)
        ps.setLong(10,new Date().getTime)
        ps.setString(11,MD5Utils.getMd5String(crawlerTask))
        ps.addBatch()
      })
      ps.executeBatch()
      connectionPool.setAutoCommit(true)
    } catch {
      case e:Exception => e.printStackTrace()
    } finally {
      ps.close()
      connectionPool.close()
    }
  }

  override def update(crawlerTask: TaskRow): Unit = {
    val connectionPool = Datasource.getConnectionPool(crawlerTask.source).getConnection
    connectionPool.setAutoCommit(false)
    var stmt: Statement = null
    try {
      stmt = connectionPool.createStatement()
      val update = s"update task set status = ${crawlerTask.status} where md5 = '${crawlerTask.md5}'"
      stmt.execute(update)
      connectionPool.setAutoCommit(true)
    } catch {
      case e:Exception => e.printStackTrace()
    } finally {
      stmt.close()
      connectionPool.close()
    }
  }

  override def delete(crawlerTask: TaskRow): Unit = {

  }

  override def find(crawlerTask: TaskRow): List[TaskParamMessage] = {
    val taskList = new collection.mutable.ListBuffer[TaskParamMessage]()
    val connectionPool = Datasource.getConnectionPool(crawlerTask.source).getConnection

    connectionPool.setAutoCommit(false)
    val sql = """create table IF NOT EXISTS "TASK"(
      "METHOD" VARCHAR NOT NULL ,
      "DATA" VARCHAR NOT NULL ,
      "HOST" VARCHAR NOT NULL ,
      "PARAM" VARCHAR NOT NULL ,
      "HEADER" VARCHAR NOT NULL ,
      "SOURCE" VARCHAR NOT NULL,
      "TYPE"  VARCHAR NOT NULL,
      "STATUS" INT NOT NULL,
      "PRIOR" INT NOT NULL,
      "DATE" BIGINT NOT NULL,
      "MD5" VARCHAR NOT NULL
    );
    CREATE INDEX IF NOT EXISTS ind_md5 ON  TASK(MD5);"""
    connectionPool.createStatement().execute(sql)

    var stmt: Statement = null
    try {
      stmt = connectionPool.createStatement()
      val select =s"select * from task where md5 = '${MD5Utils.getMd5String((crawlerTask.host+crawlerTask.data+crawlerTask.param))}'"
      val rs = stmt.executeQuery(select)
      while (rs.next()){
        taskList.append(TaskParamMessage(rs.getString(1),rs.getString(2),rs.getString(3),rs.getString(4),rs.getString(5),rs.getString(6),
          rs.getString(7),rs.getInt(8),rs.getInt(9)))
      }
      log.info(s"有${taskList.length}个任务被分配")
    } catch {
      case e:Exception => e.printStackTrace()
    } finally {
      stmt.close()
      connectionPool.close()
    }
    taskList.toList
  }

  override def findByPrior(dbName:String): List[TaskParamMessage] = {
    val taskList = new collection.mutable.ListBuffer[TaskParamMessage]()
    val connectionPool = Datasource.getConnectionPool(dbName).getConnection
    var stmt: Statement = null
    try {
      stmt = connectionPool.createStatement()
      val select ="select * from task where status = 0 order by prior desc limit 5"
      val rs = stmt.executeQuery(select)
      while (rs.next()){
        taskList.append(TaskParamMessage(rs.getString(1),rs.getString(2),rs.getString(3),rs.getString(4),rs.getString(5),rs.getString(6),
          rs.getString(7),rs.getInt(8),rs.getInt(9)))
      }
      log.info(s"有${taskList.length}个任务被分配")
    } catch {
      case e:Exception => e.printStackTrace()
    } finally {
      stmt.close()
      connectionPool.close()
    }
    if (taskList.size > 0){
      this.batchUpdate(taskList.map(task => TaskParamMessage(task.method,task.data,task.host,task.param,task.header,task.source,task.`type`,2,task.prior)).toList)
    }

    taskList.toList
  }

  override def findTimeOutTask(dbName: String): List[TaskParamMessage] = {
    val taskList = new collection.mutable.ListBuffer[TaskParamMessage]()
    val connectionPool = Datasource.getConnectionPool(dbName).getConnection
    var stmt: Statement = null
    try {
      stmt = connectionPool.createStatement()
      val dateTime = new Date().getTime - 1000*20*60
      val select =s"select * from task where status = 2 and date < ${dateTime} order by prior desc limit 5"
      val rs = stmt.executeQuery(select)
      while (rs.next()){
        taskList.append(TaskParamMessage(rs.getString(1),rs.getString(2),rs.getString(3),rs.getString(4),rs.getString(5),rs.getString(6),
          rs.getString(7),rs.getInt(8),rs.getInt(9)))
      }
    } catch {
      case e:Exception => e.printStackTrace()
    } finally {
      stmt.close()
      connectionPool.close()
    }

    taskList.toList
  }

  override def findRunningTasks(dbName: String): Int = {
    var runningCount = 0
    val connectionPool = Datasource.getConnectionPool(dbName).getConnection
    var stmt: Statement = null
    try {
      stmt = connectionPool.createStatement()
      val select = "select count(*) from task where status = 2"
      val st = stmt.executeQuery(select)
      while (st.next()){
        runningCount = st.getInt(1)
      }
    } catch {
      case e:Exception => e.printStackTrace()
    } finally {
      stmt.close()
      connectionPool.close()
    }
    runningCount
  }

  override def findTasksGroup(dbName: String): List[Tuple2[Int, Long]] = {
    val taskList = new collection.mutable.ListBuffer[Tuple2[Int, Long]]()
    val connectionPool = Datasource.getConnectionPool(dbName).getConnection
    var stmt: Statement = null
    try {
      stmt = connectionPool.createStatement()
      val select = "select count(*), status from task group by status"
      val st = stmt.executeQuery(select)
      while (st.next()){
        taskList.append(Tuple2(st.getInt(2),st.getLong(1)))
      }
    } catch {
      case e:Exception => e.printStackTrace()
    } finally {
      stmt.close()
      connectionPool.close()
    }
    taskList.toList
  }

  override def findByMd5(md5List: List[String], dbName: String): List[TaskParamMessage] = {
    val taskList = new collection.mutable.ListBuffer[TaskParamMessage]()
    val connectionPool = Datasource.getConnectionPool(dbName).getConnection
    var stmt: Statement = null
    try {
      stmt = connectionPool.createStatement()
      val select =s"select * from task where md5 in ('${md5List.mkString("','")}}')"
      val rs = stmt.executeQuery(select)
      while (rs.next()){
        taskList.append(TaskParamMessage(rs.getString(1),rs.getString(2),rs.getString(3),rs.getString(4),rs.getString(5),rs.getString(6),
          rs.getString(7),rs.getInt(8),rs.getInt(9)))
      }
    } catch {
      case e:Exception => e.printStackTrace()
    } finally {
      stmt.close()
      connectionPool.close()
    }

    taskList.toList
  }

  def taskRowToTaskParam(tasks: List[TaskRow]): List[TaskParamMessage] = {
    val vec = new ListBuffer[TaskParamMessage]
    for (task <- tasks) {
        vec += TaskParamMessage(task.method,task.data,task.host,task.param,task.header,task.source,task.`type`,task.status,task.prior)
    }
    vec.toList
  }

  def updateTaskStatus(crawlerTask: TaskRow): Unit ={
    val connectionPool = Datasource.getConnectionPool(crawlerTask.source).getConnection
    connectionPool.setAutoCommit(false)
    var stmt: Statement = null
    try {
      stmt = connectionPool.createStatement()
      val update = s"update task set status = ${crawlerTask.status} where md5 = '${crawlerTask.md5}'"
      stmt.execute(update)
      connectionPool.setAutoCommit(true)
    } catch {
      case e:Exception => e.printStackTrace()
    } finally {
      stmt.close()
      connectionPool.close()
    }
  }

  override def batchUpdate(crawlerTasks: List[TaskParamMessage]): Unit = {
    val connectionPool = Datasource.getConnectionPool(crawlerTasks(0).source).getConnection
    connectionPool.setAutoCommit(false)
    var stmt: Statement = null
    try {
      stmt = connectionPool.createStatement()
      var sql = s"update task set status = case md5 "
      var md5s = List[String]()
//      var hostList = List[String]()
      crawlerTasks.map(crawlerTask => {
        sql = sql + s" when '${MD5Utils.getMd5String((crawlerTask.host+crawlerTask.data+crawlerTask.param))}' then ${crawlerTask.status}"
        md5s = MD5Utils.getMd5String((crawlerTask.host+crawlerTask.data+crawlerTask.param)) :: md5s
//        sql = sql + s" when '${crawlerTask.host}' then ${crawlerTask.status}"
//        hostList = crawlerTask.host :: hostList
      })
      sql = sql + s" end, date = '${new Date().getTime}' where md5 in ('${md5s.mkString("','")}')"
//      println(sql)
      stmt.execute(sql)
      connectionPool.setAutoCommit(true)
    } catch {
      case e:Exception => e.printStackTrace()
    } finally {
      stmt.close()
      connectionPool.close()
    }
  }
}